package study.springboot.kafka.consumer.support.kafka;

import lombok.extern.slf4j.Slf4j;
import org.springframework.boot.autoconfigure.kafka.KafkaProperties;
import org.springframework.boot.context.properties.ConfigurationProperties;
import org.springframework.context.annotation.Bean;
import org.springframework.context.annotation.Configuration;
import org.springframework.kafka.config.ConcurrentKafkaListenerContainerFactory;
import org.springframework.kafka.config.KafkaListenerContainerFactory;
import org.springframework.kafka.core.ConsumerFactory;
import org.springframework.kafka.core.DefaultKafkaConsumerFactory;
import org.springframework.kafka.listener.ConcurrentMessageListenerContainer;
import org.springframework.kafka.listener.ContainerProperties;

@Slf4j
@Configuration
public class KafkaListenerContainer_02_Config {

    @Bean
    @ConfigurationProperties(prefix = "kafka.k2")
    public KafkaProperties kafkaProperties() {
        return new KafkaProperties();
    }

    @Bean
    public ConsumerFactory<String, String> consumerFactory(KafkaProperties kafkaProps) {
        ConsumerFactory factory = new DefaultKafkaConsumerFactory<>(kafkaProps.buildConsumerProperties());
        return factory;
    }

    @Bean(name = "kafkaListenerContainerFactory_02")
    public KafkaListenerContainerFactory<ConcurrentMessageListenerContainer<String, String>> kafkaListenerContainerFactory(ConsumerFactory consumerFactory) {
        ConcurrentKafkaListenerContainerFactory<String, String> factory = new ConcurrentKafkaListenerContainerFactory<>();

        //
        factory.setConsumerFactory(consumerFactory);
        //并发数 多个微服务实例会均分
        factory.setConcurrency(2);
        //批次监听器
        factory.setBatchListener(true);
        //容器属性
        ContainerProperties containerProps = factory.getContainerProperties();
        //是否设置手动提交
        containerProps.setAckMode(ContainerProperties.AckMode.MANUAL_IMMEDIATE);
        containerProps.setAckCount(10);
        containerProps.setAckTime(1000);

        return factory;
    }

}
